package com.bfxy.rabbit.producer.broker;

import com.bfxy.rabbit.api.Message;
import com.bfxy.rabbit.api.MessageType;
import com.bfxy.rabbit.producer.constant.BrokerMessageConst;
import com.bfxy.rabbit.producer.constant.BrokerMessageStatus;
import com.bfxy.rabbit.producer.entity.BrokerMessage;
import com.bfxy.rabbit.producer.service.MessageStoreService;
import com.sun.corba.se.pept.broker.Broker;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.DateUtils;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;

/**
 * @describte:TODO
 * @program:rabbit-parent
 * @author:zhengdp
 * @create:2021-09-14 21:42:08
 **/
@Slf4j
@Component
public class RabbitBrokerImpl implements RabbitBroker{

    @Autowired
    private RabbitTemplateContainer rabbitTemplateContainer;

    @Autowired
    private MessageStoreService messageStoreService;

    /**
     * 可靠性投递
     * 在消息投递的时候要将其存储到数据库中
     *
     * @param message
     */
    @Override
    public void reliantSend(Message message) {
        message.setMessageType(MessageType.Reliant);
        // 判断一下数据库中是否有该消息记录
        BrokerMessage msg = messageStoreService.query(message.getMessageId());
        if(msg == null){
            // 说明是第一次发送 可靠消息
            // 1. 把数据库的消息发送日志先记录好
            Date now = new Date();
            BrokerMessage brokerMessage = new BrokerMessage();
            brokerMessage.setMessage(message);
            brokerMessage.setMessageId(message.getMessageId());
            brokerMessage.setStatus(BrokerMessageStatus.SENDING.getCode());
            // tryCount 第一次发送失败，第二次才算是重试一次
            // 所以trycount在最开始发送的时候不需要进行设置
            // 下次重试时间，当前时间 + 间隔时间
            brokerMessage.setNextRetry(DateUtils.addMilliseconds(now, BrokerMessageConst.TIMEOUT));
            brokerMessage.setCreateTime(now);
            brokerMessage.setUpdateTime(now);
            // 将 BrokerMessage入库
            messageStoreService.insert(brokerMessage);
        }
        // 2. 执行真正的发送消息逻辑
        sendKernel(message);

    }

    /**
     * 迅速发送消息
     * @param message
     */
    @Override
    public void rapidSend(Message message) {
        message.setMessageType(MessageType.RAPID);
        sendKernel(message);
    }

    @Override
    public void confirmSend(Message message) {
        message.setMessageType(MessageType.CONFIRM);
        sendKernel(message);
    }



    /*
    批量发送消息,其实也是一条条的发
     */
    @Override
    public void sendMessages() {
        // 从当前线程中取到所有数据
        List<Message> messages = MessageHolder.clear();
        messages.forEach(message -> {
            MessageHolderAsyncQueue.submit(() -> {
                // CorrelationData 带上 消息类型，方便之后的confirm来判断消息类型
                CorrelationData correlationData =
                        new CorrelationData(String.format("%s#%s#%s",
                                message.getMessageId(),
                                System.currentTimeMillis(),
                                message.getMessageType())
                        );
                String topic = message.getTopic();
                String routingKey = message.getRoutingKey();
                RabbitTemplate rabbitTemplate = rabbitTemplateContainer.getTemplate(message);
                rabbitTemplate.convertAndSend(topic,
                        routingKey,message,correlationData);
                log.info("#RabbitBrokerImpl.sendMessages# send to rabbitmq,messageId: {}",message.getMessageId());
            });
        });
    }

    /**
     * 发送消息的核心方法，使用异步线程池发送消息
     * 使用SpringBoot去发送消息，使用RabbitTemplate来进行发送
     * 迅速消息直接扔到mq即可，不需要其他的操作
     * @param message
     */
    private void sendKernel(Message message){

        AsyncQueue.submit(() -> {
            // CorrelationData 带上 消息类型，方便之后的confirm来判断消息类型
            CorrelationData correlationData =
                    new CorrelationData(String.format("%s#%s#%s",
                            message.getMessageId(),
                            System.currentTimeMillis(),
                            message.getMessageType())
                            );
            String topic = message.getTopic();
            String routingKey = message.getRoutingKey();
            RabbitTemplate rabbitTemplate = rabbitTemplateContainer.getTemplate(message);
            rabbitTemplate.convertAndSend(topic,
                    routingKey,message,correlationData);
            log.info("#RabbitBrokerImpl.sendKernel# send to rabbitmq,messageId: {}",message.getMessageId());
        });
    }
}
